package com.mjf.spark.day06

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 检查点
 *    1）检查点：是通过将RDD中间结果写入磁盘。
 *    2）为什么要做检查点？
 *    由于血缘依赖过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果检查点之后有节点出现问题，可以从检查点开始重做血缘，减少了开销。
 *    3）检查点存储路径：Checkpoint的数据通常是存储在HDFS等容错、高可用的文件系统
 *    4）检查点数据存储格式为：二进制的文件
 *    5）检查点切断血缘：在Checkpoint的过程中，该RDD的所有依赖于父RDD中的信息将全部被移除。
 *    6）检查点触发时间：对RDD进行checkpoint操作并不会马上被执行，必须执行Action操作才能触发。但是检查点为了数据安全，会从血缘关系的最开始执行一遍
 */
object Spark02_CheckPoint {
  def main(args: Array[String]): Unit = {

    // 创建SparkConf配置文件
    val conf = new SparkConf().setMaster("local[*]").setAppName("Spark02_CheckPoint")
    // 创建SparkContext对象
    val sc = new SparkContext(conf)

    // 设置检查点目录
    sc.setCheckpointDir("D:\\code\\spark\\cp")

    // 开发环境，应该将检查点目录设置在hdfs上
    // 设置访问HDFS集群的用户名
//    System.setProperty("HADOOP_USER_NAME","mjf")
//    sc.setCheckpointDir("hdfs://hadoop102:9000/cp")


    val rdd: RDD[String] = sc.makeRDD(List("hello lucy", "hello jack"), 2)

    val flatMapRDD: RDD[String] = rdd.flatMap(_.split(" "))

    val mapRDD: RDD[(String, Long)] = flatMapRDD.map(
      word => {
        (word, System.currentTimeMillis())
      }
    )

    // 打印血缘关系
    println(mapRDD.toDebugString)

    // 在开发环境，一般检查点和缓存配合使用
    mapRDD.cache()

    // 设置检查点
    mapRDD.checkpoint()

    // 触发行动操作
    mapRDD.collect().foreach(println)

    println("------------------------------")

    // 打印血缘关系
    println(mapRDD.toDebugString)
    // 触发行动操作
    mapRDD.collect().foreach(println)

    // 释放缓存
    mapRDD.unpersist()

    // 关闭连接
    sc. stop()

  }
}
